package com.hu.flink12.sensor

import java.util.Random

import com.hu.entity.SensorReading
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

/**
 * @Author: 22834
 * @Date: 2020/11/13 23:05
 * @Describe: 自定义读取数据源
 */

object ReadSensorDefineSource {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //读取自定义数据源
    val mySensorSourceStream = env.addSource(new MySensorSource())
    mySensorSourceStream.keyBy(new KeySelector[SensorReading,String]{
      override def getKey(in: SensorReading): String = {
        in.id
      }
    }).sum("temperature").print().setParallelism(1)

    env.execute()
  }
}

class MySensorSource() extends SourceFunction[SensorReading] {

  var runFlag = true

  override def run(sourceContext: SourceFunction.SourceContext[SensorReading]): Unit = {
    val random = new Random()
    var curData = 1.to(10).map(i => ("sensor_" + i, random.nextDouble() * 100))

    while (runFlag) {
      curData = curData.map(data => (data._1, data._2 + random.nextGaussian()))

      curData.foreach(data =>
        sourceContext.collect(SensorReading(data._1.trim, System.currentTimeMillis(), data._2)))

      Thread.sleep(500)
    }
  }

  override def cancel(): Unit = runFlag = false
}
